package simpledb.execution;

import simpledb.common.DbException;
import simpledb.common.Type;
import simpledb.storage.*;
import simpledb.transaction.TransactionAbortedException;

import java.util.*;

/**
 * Knows how to compute some aggregate over a set of IntFields.
 */
public class IntegerAggregator implements Aggregator {

    // 与IntegerAggregator实例字段不绑定，设为静态内部类，减少空间浪费，加快垃圾回收
    // 统计每一个组的信息
    private static class AggregateState {
        int sum;    // 用于SUM、AVG
        int count;  // 用于COUNT、AVG
        int min;    // 用于MIN
        int max;    // 用于MAX

        // 初始化聚合状态（第一次遇到该分组时）
        public AggregateState(int initialVal) {
            this.sum = initialVal;
            this.count = 1;
            this.min = initialVal;
            this.max = initialVal;
        }
    }

    private static final long serialVersionUID = 1L;

    private final int gbfield;
    private final Type gbfieldtype;
    private final int afield;
    private final Op what;
    // 存储分组到聚合状态的映射：key为分组字段值（无分组时为null），value为聚合中间结果
    private Map<Field, AggregateState> groupMap;

    /**
     * Aggregate constructor
     * 
     * @param gbfield
     *            the 0-based index of the group-by field in the tuple, or
     *            NO_GROUPING if there is no grouping
     * @param gbfieldtype
     *            the type of the group by field (e.g., Type.INT_TYPE), or null
     *            if there is no grouping
     * @param afield
     *            the 0-based index of the aggregate field in the tuple
     * @param what
     *            the aggregation operator
     *
     * 按哪个字段分组（gbfield） 按部门分组
     * 分组字段的类型（gbfieldtype）
     * 对哪个字段进行聚合计算（afield）对工资字段聚合，聚合字段固定，本类处理int
     * 具体执行哪种聚合操作（what）
     *
     */

    public IntegerAggregator(int gbfield, Type gbfieldtype, int afield, Op what) {
        // some code goes here
        this.gbfield = gbfield;
        this.gbfieldtype = gbfieldtype;
        this.afield = afield;
        this.what = what;
        // 使用hashmap处理无分组key为null的情况
        this.groupMap = new HashMap<>();
    }

    /**
     * Merge a new tuple into the aggregate, grouping as indicated in the
     * constructor
     * 
     * @param tup
     *            the Tuple containing an aggregate field and a group-by field
     */
    public void mergeTupleIntoGroup(Tuple tup) {
        // some code goes here
        Field groupKey;
        if (gbfield == Aggregator.NO_GROUPING) {
            groupKey = null;
        }else {
            groupKey = tup.getField(gbfield);
        }

        // 提取聚合字段的值，并添加类型校验, 必须是IntField
        Field field = tup.getField(afield);
        if (!(field instanceof IntField)) {
            throw new RuntimeException("IntegerAggregator requires aggregate field to be IntField, but got " + field.getClass().getSimpleName());
        }

        int val = ((IntField) field).getValue();

        // 分组
        AggregateState state = groupMap.get(groupKey);
        if (state==null) {
            groupMap.put(groupKey, new AggregateState(val));
        }else {
            switch (what) {
                // 箭头不会穿透，但多行命令需要加大括号，冒号不用
                case MIN -> {
                    state.min = Math.min(val, state.min);
                }
                case MAX -> {
                    state.max = Math.max(val, state.max);
                }
                case COUNT -> {
                    state.count++;
                }
                // SUM额外维护count，统一操作
                case SUM, AVG, SC_AVG, SUM_COUNT -> {
                    state.count++;
                    state.sum += val;
                    // System.out.println(groupKey.toString() + ": " +state.sum);
                }
            }
        }
    }

    private int getResult(AggregateState state) {
        switch (what) {
            case COUNT -> {
                return state.count;
            }
            case SUM -> {
                return state.sum;
            }
            case AVG -> {
                return state.sum / state.count;  // 整数除法
            }
            case MIN -> {
                return state.min;
            }
            case MAX -> {
                return state.max;
            }
            default -> {
                throw new RuntimeException("op 不合法" + what);
            }
        }
    }

    // 非静态，需要访问外部类的字段
    private class AggregateIterator implements OpIterator {
        private final TupleDesc td;
        private List<Tuple> resultTuples;
        // 无需和文件关联所以不用DbFileIterator
        private Iterator<Tuple> iterator;


        public AggregateIterator() {
            this.resultTuples = new ArrayList<>();
            if (IntegerAggregator.this.gbfield == Aggregator.NO_GROUPING) {
                this.td = new TupleDesc(new Type[]{Type.INT_TYPE});
                Tuple tuple = new Tuple(this.td);
                tuple.setField(0, new IntField(getResult(groupMap.get(null))));
                resultTuples.add(tuple);
            }else {
                this.td = new TupleDesc(new Type[]{IntegerAggregator.this.gbfieldtype, Type.INT_TYPE});
                for (Map.Entry<Field, AggregateState> entry : IntegerAggregator.this.groupMap.entrySet()) {
                    Tuple tuple = new Tuple(this.td);
                    tuple.setField(0, entry.getKey());
                    tuple.setField(1, new IntField(getResult(groupMap.get(entry.getKey()))));
                    resultTuples.add(tuple);
                }
            }
            this.iterator = null;
        }

        /**
         * Opens the iterator. This must be called before any of the other methods.
         *
         * @throws DbException when there are problems opening/accessing the database.
         */
        @Override
        public void open() throws DbException, TransactionAbortedException {
            iterator = resultTuples.iterator();
        }

        /**
         * Returns true if the iterator has more tuples.
         *
         * @return true f the iterator has more tuples.
         * @throws IllegalStateException If the iterator has not been opened
         */
        @Override
        public boolean hasNext() throws DbException, TransactionAbortedException {
            if (iterator==null) {
                throw new IllegalStateException("iterator尚未初始化");
            }
            return iterator.hasNext();
        }

        /**
         * Returns the next tuple from the operator (typically implementing by reading
         * from a child operator or an access method).
         *
         * @return the next tuple in the iteration.
         * @throws NoSuchElementException if there are no more tuples.
         * @throws IllegalStateException  If the iterator has not been opened
         */
        @Override
        public Tuple next() throws DbException, TransactionAbortedException, NoSuchElementException {
            if (iterator==null) {
                throw new IllegalStateException("iterator尚未初始化");
            }
            if (!iterator.hasNext()) {
                throw new NoSuchElementException("there are no more tuples");
            }
            return iterator.next();
        }

        /**
         * Resets the iterator to the start.
         *
         * @throws DbException           when rewind is unsupported.
         * @throws IllegalStateException If the iterator has not been opened
         */
        @Override
        public void rewind() throws DbException, TransactionAbortedException {
            if (iterator==null) {
                throw new IllegalStateException("iterator尚未初始化");
            }
            open();
        }

        /**
         * Returns the TupleDesc associated with this OpIterator.
         *
         * @return the TupleDesc associated with this OpIterator.
         */
        @Override
        public TupleDesc getTupleDesc() {
            return td;
        }

        /**
         * Closes the iterator. When the iterator is closed, calling next(),
         * hasNext(), or rewind() should fail by throwing IllegalStateException.
         */
        @Override
        public void close() {
            iterator = null;
        }
    }

    /**
     * Create a OpIterator over group aggregate results.
     * 
     * @return a OpIterator whose tuples are the pair (groupVal, aggregateVal)
     *         if using group, or a single (aggregateVal) if no grouping. The
     *         aggregateVal is determined by the type of aggregate specified in
     *         the constructor.
     */
    public OpIterator iterator() {
        // some code goes here
        return new AggregateIterator();
    }


}
